package com.alibaba.otter.canal.instance.spring;

import java.util.List;

import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.instance.core.CanalMQConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.model.Event;

/**
 * 基于spring容器启动canal实例，方便独立于manager启动
 * CanalInstanceWithSpring是AbstractCanalInstance的子类，提供了一些set方法为instance的组成模块赋值
 *
 * 当我们配置加载方式为spring时，创建的CanalInstance实例类型都是CanalInstanceWithSpring。
 * canal将会寻找本地的spring配置文件来创建instance实例。
 *
 * canal默认提供了一下几种spring配置文件：
 *     spring/memory-instance.xml
 *         所有的组件(parser , sink , store)都选择了内存版模式，记录位点的都选择了memory模式，重启后又会回到初始位点进行解析
 *         特点：速度最快，依赖最少(不需要zookeeper)
 *         场景：一般应用在quickstart，或者是出现问题后，进行数据分析的场景，不应该将其应用于生产环境
 *     spring/file-instance.xml
 *         所有的组件(parser , sink , store)都选择了基于file持久化模式，注意，不支持HA机制.
 *         特点：支持单机持久化
 *         场景：生产环境，无HA需求，简单可用.
 *     spring/default-instance.xml
 *         所有的组件(parser , sink , store)都选择了持久化模式，目前持久化的方式主要是写入zookeeper，保证数据集群共享.
 *         特点：支持HA
 *         场景：生产环境，集群化部署.
 *     spring/group-instance.xml
 *         主要针对需要进行多库合并时，可以将多个物理instance合并为一个逻辑instance，提供客户端访问。
 *         场景：分库业务。
 *         比如产品数据拆分了4个库，每个库会有一个instance，如果不用group，业务上要消费数据时，需要启动4个客户端，分别链接4个instance实例。
 *         使用group后，可以在canal server上合并为一个逻辑instance，只需要启动1个客户端，链接这个逻辑instance即可.
 *
 * 这几个不同的spring配置文件中，最主要的就是metaManager 和eventParser 这两个配置有所不同，eventStore、和eventSink 定义都是相同的。这是因为：
 *     eventStore：目前的开源版本中eventStore只有一种基于内存的实现，所以配置都相同
 *     eventSink：其作用是eventParser和eventStore的链接器，进行数据过滤，加工，分发的工作。不涉及存储，也就没有必要针对内存、file、或者zk进行区分。
 *
 * @author jianghang 2012-7-12 下午01:21:26
 * @author zebin.xuzb
 * @version 1.0.0
 */
public class CanalInstanceWithSpring extends AbstractCanalInstance {

    private static final Logger logger = LoggerFactory.getLogger(CanalInstanceWithSpring.class);

    @Override
    public void start() {
        logger.info("start CannalInstance for {}-{} ", new Object[] { 1, destination });
        super.start();
    }

    // ======== setter ========

    public void setDestination(String destination) {
        this.destination = destination;
    }

    public void setEventParser(CanalEventParser eventParser) {
        this.eventParser = eventParser;
    }

    public void setEventSink(CanalEventSink<List<CanalEntry.Entry>> eventSink) {
        this.eventSink = eventSink;
    }

    public void setEventStore(CanalEventStore<Event> eventStore) {
        this.eventStore = eventStore;
    }

    public void setMetaManager(CanalMetaManager metaManager) {
        this.metaManager = metaManager;
    }

    public void setAlarmHandler(CanalAlarmHandler alarmHandler) {
        this.alarmHandler = alarmHandler;
    }

    public void setMqConfig(CanalMQConfig mqConfig){
        this.mqConfig = mqConfig;
    }

}
